/*
 * Copyright (C), 2010-2017, ATCOM Tech. Co., Ltd.
 *
 * File Name: threadpool.c
 * Author: liaoxuecheng
 * mail: 872958436@qq.com
 * Created Time: Tue 01 Sep 2020 08:53:10 AM CST
 *
 */

#include<stdio.h>
#include<string.h>
#include<stdlib.h>
#include<unistd.h>
#include <sys/types.h>
#include <sys/stat.h>
#include <fcntl.h>
#include <pthread.h>

#include "threadpool.h"
#include "threadlog.h"

static char const __invoke_dynamic_linker__[] __attribute__ ((section (".interp")))
#ifdef __LP64__
= "/lib64/ld-linux-x86-64.so.2";
#else
= "/lib/ld-linux.so.2";
#endif

void _start(void)
{
    static char const msg[] = "This is a threadpool library!\n";
    write(STDOUT_FILENO, msg, sizeof(msg) - 1);
    _exit(0);
}

#define THIS_FILE "threadpool.c"

#define CLOSE_IMMEDIATELY 1
#define CLOSE_GRACEFULLY 2

static void *threadpool_func(void *arg)
{

    threadpool_t *pool = (threadpool_t *)arg;
    threadpool_task_t task;
    while(1)
    {
        pthread_mutex_lock(&(pool->mutex));
        /* 队列为空时，就等待队列非空 */
        while(pool->queue_cur_num == 0 && !pool->pool_close)
        {
            DEBUG_INFO(THIS_FILE, "Queue empty now, waiting!\n");
            pthread_cond_wait(&(pool->queue_not_empty), &(pool->mutex));
        }
        if(pool->pool_close == CLOSE_IMMEDIATELY || (pool->pool_close == CLOSE_GRACEFULLY && pool->queue_cur_num == 0))
        {
            DEBUG_NOTICE(THIS_FILE, "threadpool exit now, thread id: %lx\n", pthread_self());
            pthread_mutex_unlock(&(pool->mutex));
            pthread_exit(NULL);
        }

        task.func = pool->queue[pool->head].func;
        task.arg = pool->queue[pool->head].arg;
        pool->head = (pool->head + 1) % pool->queue_max_num;
        pool->queue_cur_num -= 1;

        /* 队列为空, 发送通知, 可以销毁线程函数 */
        if(pool->queue_cur_num == 0)
        {
            DEBUG_INFO(THIS_FILE, "Queue empty now, sending queue_empty signal!\n");
            pthread_cond_signal(&(pool->queue_empty));
        }
        /* 队列非满, 发送通知, 可以添加新任务 */
        else if(pool->queue_cur_num == pool->queue_max_num - 1)
        {
            DEBUG_INFO(THIS_FILE, "Queue is not full now, sending queue_not_full signal!\n");
            pthread_cond_broadcast(&(pool->queue_not_full));
        }
        pthread_mutex_unlock(&(pool->mutex));
        /* 线程真正要做的工作, 回调函数的调用 */
        task.func(task.arg);
    }
}

int threadpool_destory(threadpool_t *pool)
{
    int i, flags;
    if(pool == NULL)
        return THREADPOOL_INVALID;

    pthread_mutex_lock(&(pool->mutex));

    if(pool->pool_close)
    {
        DEBUG_WARN(THIS_FILE, "Threadpool is closing now, wait a moment!\n");
        pthread_mutex_unlock(&(pool->mutex));
        return THREADPOOL_SHUTDOWN;
    }

    if(pool->flags & THREADPOOL_CLOSE_IMMEDIATELY || pool->flags & THREADPOOL_CLOSE_IMDY_ANDABORT)
        flags = CLOSE_IMMEDIATELY;
    else
        flags = CLOSE_GRACEFULLY;

    pool->pool_close = flags;

    if(pool->pool_close == CLOSE_IMMEDIATELY)
    {
        DEBUG_NOTICE(THIS_FILE, "Threadpool is close immediately, queue func will not execute!\n");
    }

    if(pool->flags & THREADPOOL_CLOSE_IMDY_ANDABORT)
    {
        for(i = 0; i < pool->thread_count; i++)
        {
            pthread_cancel(pool->threads[i]);
        }
    }

    /* 等待队列为空 */
    while(pool->pool_close == CLOSE_GRACEFULLY && pool->queue_cur_num != 0)
    {
        DEBUG_INFO(THIS_FILE, "Threadpool is gracefully close, waiting queue_empty signal!\n");
        pthread_cond_wait(&(pool->queue_empty), &(pool->mutex));
    }

    pthread_mutex_unlock(&(pool->mutex));
    pthread_cond_broadcast(&(pool->queue_not_empty));
    pthread_cond_broadcast(&(pool->queue_not_full));

    for(i = 0; i < pool->thread_count; i++)
    {
        DEBUG_NOTICE(THIS_FILE, "pthread_join now, thread id: %lx!\n", pool->threads[i]);
        pthread_join(pool->threads[i], NULL);
    }

    pthread_mutex_destroy(&(pool->mutex));
    pthread_cond_destroy(&(pool->queue_empty));
    pthread_cond_destroy(&(pool->queue_not_empty));
    pthread_cond_destroy(&(pool->queue_not_full));

    if(pool->threads)
    {
        free(pool->threads);
        free(pool->queue);
    }
    free(pool);
    DEBUG_WARN(THIS_FILE, "%s success!\n", __func__);
}

int threadpool_create(threadpool_t **pool, int thread_count, int queue_size, int flags)
{
    pthread_attr_t attr;
    int stack_size = 81920;

    int limit_queue_size = MIN_QUEUE_SIZE;
    int limit_thread_count = MIN_THREADS;
    int err = THREADPOOL_CREATE_SUCCESS;

    if(thread_count <= 0)
        thread_count = MAX_THREADS;
    if(queue_size <= 0)
        queue_size = MAX_QUEUE_SIZE;

    if(pool == NULL || thread_count <= 0 || queue_size <= 0)
    {
        err = THREADPOOL_CREATE_PARAM_ERR;
        goto ERR;
    }

    *pool = (threadpool_t *)malloc(sizeof(threadpool_t));

    if(*pool == NULL)
    {
        err = THREADPOOL_MALLOC_ERR;
        goto ERR;
    }
    memset(*pool, 0, sizeof(threadpool_t));

    if(flags & THREADPOOL_MAX_QUEUE_SIZE)
        limit_queue_size = MAX_QUEUE_SIZE;
    else if(flags & THREADPOOL_MIDDLE_QUEUE_SIZE)
        limit_queue_size = MIDDLE_QUEUE_SIZE;

    if(flags & THREADPOOL_MAX_THREADS)
        limit_thread_count = MAX_THREADS;
    else if(flags & THREADPOOL_MIDDLE_THREADS)
            limit_thread_count = MIDDLE_THREADS;

    if(queue_size > limit_queue_size)
        queue_size = limit_queue_size;
    if(thread_count > limit_thread_count)
        thread_count = limit_thread_count;

    (*pool)->threads = (pthread_t *)malloc(sizeof(pthread_t) * thread_count);
    (*pool)->queue = (threadpool_task_t *)malloc(sizeof(threadpool_task_t) * queue_size);
    (*pool)->thread_count = thread_count;
    (*pool)->queue_max_num = queue_size;
    (*pool)->flags = flags;

    if(NULL == (*pool)->threads || NULL == (*pool)->queue)
    {
        err = THREADPOOL_MALLOC_ERR;
        goto ERR;
    }

    pthread_mutex_init(&((*pool)->mutex), NULL);
    pthread_cond_init(&((*pool)->queue_empty), NULL);
    pthread_cond_init(&((*pool)->queue_not_empty), NULL);
    pthread_cond_init(&((*pool)->queue_not_full), NULL);

    pthread_attr_init(&attr);
    pthread_attr_setstacksize(&attr, stack_size);
    int i;
    for(i = 0; i < thread_count; i++)
    {
        if(pthread_create(&((*pool)->threads[i]), &attr, threadpool_func, (void *)*pool) != 0)
        {
            err = THREADPOOL_CREATE_THREAD_ERR;
            pthread_attr_destroy(&attr);
            goto ERR;
        }
    }
    pthread_attr_destroy(&attr);

    DEBUG_WARN(THIS_FILE, "thread_count:%d, queue_count:%d\n", thread_count, queue_size);

    return err;
ERR:
    DEBUG_ERROR(THIS_FILE, "create error, error code:%d\n", err);
    if(pool && *pool)
    {
        threadpool_destory(*pool);
        *pool = NULL;
    }
    return err;
}

int threadpool_add_job(threadpool_t *pool, void(*callback_func)(void *), void *arg, int flags)
{
    int err = 0;
    if(pool == NULL || callback_func == NULL)
    {
        return THREADPOOL_INVALID;
    }

    int next = (pool->tail + 1) % pool->queue_max_num;
    pthread_mutex_lock(&(pool->mutex));

    while ((pool->queue_cur_num == pool->queue_max_num) && !(pool->pool_close))
    {
        DEBUG_INFO(THIS_FILE, "Threadpool is full, waiting queue_not_full signal!\n");
        pthread_cond_wait(&(pool->queue_not_full), &(pool->mutex));   //队列满的时候就等待
    }

    if(pool->pool_close)
    {
        pthread_mutex_unlock(&(pool->mutex));
        return THREADPOOL_SHUTDOWN;
    }

    pool->queue[pool->tail].func = callback_func;
    pool->queue[pool->tail].arg = arg;
    pool->tail = next;
    pool->queue_cur_num += 1;
    /* 添加一个任务, 唤醒一个线程
     * pthread_cond_signal 在pthread_mutex_lock pthread_mutex_unlock 之间,或者之后都可以,
     * 但是Linux中推荐使用在之间
     * */
    pthread_cond_signal(&(pool->queue_not_empty));
    pthread_mutex_unlock(&(pool->mutex));
    DEBUG_MSG(THIS_FILE, "threadpool_add_job success! crrent cnt:%d!\n", pool->queue_cur_num);
    return 0;
}
